package net.shambolica.jport.runner;

import static net.erl.objects.ET.atom;
import static net.erl.objects.ET.number;
import static net.erl.objects.ET.tuple;
import net.erl.objects.ErlTerm;
import net.erl.objects.ErlTuple;
import static net.erl.objects.pattern.EP.ptuple;
import static net.erl.objects.pattern.EP.pvar;
import net.shambolica.jport.exception.InitializationError;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Map;
import java.util.Properties;

/**
 * I create a connection (over a socket) based a set of properties.
 * <br/>- The connection type will be chosen by the value of the property ${prefix}.type.
 * <br/>- The property ${prefix}.cookie is reserved for use by upper layer.
 * <br/>Supported connection types (and params):
 * <br/>1. TCP: (host = localhost)
 * <br/>   - ${prefix}.type = tcp
 * <br/>   - ${prefix}.port = &lt;port&gt;          [required]
 * <br/>
 */
public class ConnectionFactory {

    private static final Logger log = LoggerFactory.getLogger(ConnectionFactory.class);


    /**
     * Get a connector using a set of properties.
     * @param allProperties A set of properties, including properties relevant to
     * the set up of the required connection. These will be prefixed by the <i>prefix</i>
     * parameter.
     * @param prefix        Prefix for connection properties.
     * @return              A connector.
     * @throws net.shambolica.jport.exception.InitializationError if unable to create/initialize
     * a connector.
     * @throws java.net.ConnectException if unable to connect.
     */
    public DataConnection getConnection(Properties allProperties, String prefix)
            throws InitializationError, ConnectException {

        // Filter out relevant props.
        Properties props = getPropertiesByPrefix(allProperties, prefix, true);

        // Get conn type.
        String type = props.getProperty("type");
        if (type == null) {
            log.error("no type property found for connection initialization");
            throw new IllegalArgumentException("no type property found");
        }

        // Get cookie and expected ids.
        String cookieStr = props.getProperty("cookie");
        long cookie = cookieStr == null ? 0 : Long.parseLong(cookieStr);

        String expectedStr = props.getProperty("expected");
        long expected = expectedStr == null ? 0 : Long.parseLong(expectedStr);

        /* Create raw connection, according to props (pre-handshake). */
        DataConnection conn;
        if (type.equalsIgnoreCase("tcp")) {
            log.info("tcp connection type requested");
            // TODO: TCP connection.
            int port = props.getProperty("port") == null ? -1 : Integer.parseInt(props.getProperty("port"));
            final Socket sock = new Socket();
            try {
                sock.connect(new InetSocketAddress(InetAddress.getLocalHost(), port));
            } catch (IOException e) {
                log.error("unable to connect socket: {}: {}", e.getClass().getName(), e.getMessage());
                throw new ConnectException(e.getClass().getSimpleName()+": "+e.getMessage());
            }
            try {
                conn = new EtfStreamConnection(sock.getInputStream(), sock.getOutputStream()){
                    public synchronized void close() throws IOException {
                        super.close();
                        sock.close();
                    }
                };
            } catch (IOException e) {
                log.error("error constructing ETF stream connection: {}: {}", e, e.getMessage());
                throw new ConnectException("io exception on stream construction: "+e.getMessage());
            }
        } else {
            log.error("unknown connection type requested ({})", type);
            throw new IllegalArgumentException("unknown connection type requested ("+type+")");
        }

        /* Handshake: TX */
        try {
            log.info("sending handshake ({})", cookie);
            ErlTuple hsRequest = tuple(atom("hello"), number(cookie));
            conn.writeTerm(hsRequest);
            log.info("waiting for response ({})", expected);
            ErlTerm hsResponse = conn.readTerm();

            Map<String,ErlTerm> matchError = ptuple(atom("error"), pvar("Cause")).match(hsResponse);
            if (matchError != null) {
                log.info("got error response: {}", matchError.get("Cause"));
                throw new ConnectException("error response in handshake");
            }

        } catch (InterruptedException e) {
            if (! conn.isClosed()) {
                try {
                    conn.close();
                } catch (IOException e1) {
                    // Do nothing.
                }
            }
            throw new ConnectException(e.getClass().getName()+": "+e.getMessage());
        } catch (ConnectException e) {
            throw e;
        } catch (IOException e) {
            if (! conn.isClosed()) {
                try {
                    conn.close();
                } catch (IOException e1) {
                    // Do nothing.
                }
            }
            throw new ConnectException(e.getClass().getName()+": "+e.getMessage());
        }

        return conn;
    }


    /**
     * TODO: Move to PropertiesUtils or something.
     * Get a new properties object containing only properties prefixed by <i>prefix</i>,
     * optionally removing the prefix from the props in the new object.
     * @param properties    Original properties object.
     * @param prefix        Prefix to filter by.
     * @param trimPrefix    If true, the prefix will be trimmed (including following period).
     * @return              A new Properties Object.
     */
    public static Properties getPropertiesByPrefix(Properties properties, String prefix, boolean trimPrefix) {
        // Trim trailing period(s).
        while (prefix.endsWith(".")) {
            prefix = prefix.substring(0, prefix.length()-1).trim();
        }
        if (prefix.length() == 0) {
            return new Properties(properties);
        }
        // Add ONE trailing period, for proper matching.
        prefix = prefix+".";

        Properties ret = new Properties();
        for (Object it: properties.keySet()) {
            String key = (String)it;
            if (key.startsWith(prefix)) {
                String finalKey = trimPrefix ? key.substring(prefix.length()) : key;
                ret.put(finalKey, properties.get(key));
            }
        }

        return ret;
    }

}
